package com.winit.schedule.core.zk;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;

import com.winit.schedule.core.utils.StringUtils;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.ZooKeeper.States;
import org.apache.zookeeper.data.ACL;
import org.apache.zookeeper.data.Id;
import org.apache.zookeeper.server.auth.DigestAuthenticationProvider;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.winit.schedule.core.conf.Configuration;
import com.winit.schedule.core.conf.JobConfiguration;
import com.winit.schedule.core.service.AbstractService;

public class ZKService extends AbstractService {

    private static transient Logger log           = LoggerFactory.getLogger(ZKService.class);

    private static final String     NODE_SERVER   = "server";

    public static final String     BLACK_LIST   = "blacklist";

    private List<ClusterListener>   listeners;

    private BlackListListener       blackListListener;

    private String                  pathServer;

    private String                  pathBlackList;

    private String                  registerAddressPath;

    private String                  masterAddress = "";

    private ZooKeeper               zk;

    private List<ACL>               acl           = new ArrayList<ACL>();

    private ZKConfig                config;

    private boolean                 isMaster;

    public ZKService() throws Exception{
        this("ZKService");
    }

    public ZKService(String name) throws Exception{
        super(name);
    }

    /**
     * 重连zookeeper
     * 
     * @throws Exception
     */
    public synchronized void reConnection() throws Exception {
        if (this.zk != null) {
            log.info("---------Start connection zookeeper");
            this.zk.close();
            this.zk = null;
            this.connect();

            //addChildWatcher();
        }
    }

    @Override
    protected void serviceInit(Configuration conf) throws Exception {
        super.serviceInit(conf);
        // 设置zookeeper客户端不使用SASL认证,优先系统属性
        boolean zkSASLClient = conf.getBoolean(JobConfiguration.ZK_SASL_CLIENT, JobConfiguration.DEFAULT_ZK_SASL_CLIENT);
        System.setProperty("zookeeper.sasl.client", System.getProperty("zookeeper.sasl.client", String.valueOf(zkSASLClient)));
        
        this.config = new ZKConfig();
        config.setConnectAddr(conf.get(JobConfiguration.ZK_COONNECT_ADDR));
        config.setUsername(conf.getString(JobConfiguration.ZK_USERNAME, JobConfiguration.DEFAULT_ZK_USERNAME));
        config.setPassword(conf.getString(JobConfiguration.ZK_PASSWORD, JobConfiguration.DEFAULT_ZK_PASSWORD));
        config.setSessionTimeout(conf.getInt(JobConfiguration.ZK_SESSION_TIMEOUT,
            JobConfiguration.DEFAULT_ZK_SESSION_TIMEOUT));
        config.setRootPath(conf.getString(JobConfiguration.ZK_USCHEDULE_ROOTPATH,
            JobConfiguration.DEFAULT_ZK_USCHEDULE_ROOTPATH));
    }

    @Override
    protected void serviceStart() throws Exception {
        super.serviceStart();
        this.connect();
    }

    @Override
    protected void serviceStop() throws Exception {
        destroy();
    }

    public void destroy() throws InterruptedException {
        log.info("关闭zookeeper连接");
        this.zk.close();
    }

    private void connect() throws Exception {
        CountDownLatch connectionLatch = new CountDownLatch(1);
        createZookeeper(connectionLatch);
        connectionLatch.await();
    }

    private void createZookeeper(final CountDownLatch connectionLatch) throws Exception {
        zk = new ZooKeeper(config.getConnectAddr(), config.getSessionTimeout(), new Watcher() {

            public void process(WatchedEvent event) {
                sessionEvent(connectionLatch, event);
            }
        });
        String authString = config.getUsername() + ":" + config.getPassword();
        zk.addAuthInfo("digest", authString.getBytes());
        acl.clear();
        acl.add(new ACL(ZooDefs.Perms.ALL, new Id("digest", DigestAuthenticationProvider.generateDigest(authString))));
        acl.add(new ACL(ZooDefs.Perms.READ, Ids.ANYONE_ID_UNSAFE));
    }

    private void sessionEvent(CountDownLatch connectionLatch, WatchedEvent event) {
        if (event.getState() == KeeperState.SyncConnected) {
            log.info("Received zookeeper connection success!");
            connectionLatch.countDown();

            try {
                this.initial();
            } catch (Exception e) {
                e.printStackTrace();
            }

            if (null != listeners) {
                for (ClusterListener listener : listeners) {
                    listener.connected();
                }
            }
        } else if (event.getState() == KeeperState.Expired) {
            log.error("Session timeout, waiting for the re establishment of ZK connections...");
            try {
                reConnection();
            } catch (Exception e) {
                log.error(e.getMessage(), e);
            }
        } else if (event.getState() == KeeperState.Disconnected) {
            log.error("------------zookeeper disconnected...");
            this.masterAddress = "";
            this.isMaster = false;
            if (null != listeners) {
                for (ClusterListener listener : listeners) {
                    listener.disconnected();
                    listener.masterChanage(this.isMaster, this.masterAddress);
                }
            }
        }

        // Disconnected：Zookeeper会自动处理Disconnected状态重连
    }

    public void initial() throws Exception {
        // 当zk状态正常后才能调用
        if (zk.exists(this.getRootPath(), false) == null) {
            ZKUtils.createPath(zk, this.getRootPath(), CreateMode.PERSISTENT, acl);
            // 设置版本信息
            zk.setData(this.getRootPath(), Version.getVersion().getBytes(), -1);
        } else {
            byte[] value = zk.getData(this.getRootPath(), false, null);
            if (value == null) {
                zk.setData(this.getRootPath(), Version.getVersion().getBytes(), -1);
            } else {
                String dataVersion = new String(value);
                if (Version.isCompatible(dataVersion) == false) {
                    throw new Exception("USchedule version " + Version.getVersion()
                                        + " Incompatible version of Zookeeper " + dataVersion);
                }
                log.info("Current program version:" + Version.getVersion() + " Data version: " + dataVersion);
            }
        }

        this.pathServer = this.getRootPath() + "/" + NODE_SERVER;

        if (this.getZooKeeper().exists(this.pathServer, false) == null) {
            ZKUtils.createPath(getZooKeeper(), this.pathServer, CreateMode.PERSISTENT, this.getAcl());
        }

        if (null != blackListListener) {
            this.pathBlackList = this.getRootPath() + "/" + BLACK_LIST;

            if (this.getZooKeeper().exists(this.pathBlackList, false) == null) {
                ZKUtils.createPath(getZooKeeper(), this.pathBlackList, CreateMode.PERSISTENT, this.getAcl());
            }
        }

        addChildWatcher();

        addBlackListWatcher();
    }

    public void registerJobTracker(String host, int port) throws Exception {
        if (checkZKState()) {
            String serverId = host + ":" + port;
            registerAddressPath = ZKUtils.createJobTrackerPath(getZooKeeper(),
                this.pathServer + "/jobTracker",
                serverId.getBytes("utf-8"),
                CreateMode.EPHEMERAL_SEQUENTIAL,
                this.getAcl());
        }
    }

    public void registerJobTracker_dev(String host, int port) throws Exception {
        if (checkZKState()) {
            String serverId = host + ":" + port;
            registerAddressPath = ZKUtils.createPath_dev(getZooKeeper(),
                    this.pathServer + "/jobTracker0000000000",
                    serverId.getBytes("utf-8"),
                    CreateMode.PERSISTENT,
                    this.getAcl());
        }
    }

    public void resetBlackList(String blackList) throws Exception{
        if (StringUtils.isBlank(blackList)) {
            blackList = "";
        }
        if (StringUtils.isNotEmpty(this.pathBlackList)
            && this.getZooKeeper().exists(this.pathBlackList, false) != null) {
            byte[] data = blackList.getBytes("utf-8");
            this.getZooKeeper().setData(this.pathBlackList, data, -1);
        }
    }

    private void addBlackListWatcher() {
        try {
            log.info("------------add black watcher");

            if (null != blackListListener) {
                byte[] data = this.getZooKeeper().getData(this.pathBlackList, new Watcher() {

                    @Override public void process(WatchedEvent event) {
                        log.info("------------BlackListWatcher:BlackList Changed Event Received, EventType : " + event.getType().name());
                        if (EventType.NodeDataChanged == event.getType() || EventType.None == event.getType()) {
                            log.info("------------BlackListWatcher:BlackList Changed Event Received");

                            addBlackListWatcher();
                        }
                    }
                }, null);

                String blackList = null;
                if (null != data) {
                    blackList = new String(data, "utf-8");
                }
                log.info("-------- BlackList change : " + blackList);
                blackListListener.chanaged(blackList);
            }

        }  catch (KeeperException e) {
            log.error("addBlackListWatcher failure", e);
        } catch (InterruptedException e) {
            log.error("addBlackListWatcher failure", e);
        } catch (Exception e) {
            log.error("addBlackListWatcher failure", e);
        }

    }

    private void addChildWatcher() {
        try {
            List<String> childs = this.getZooKeeper().getChildren(this.pathServer, new Watcher() {

                @Override
                public void process(WatchedEvent event) {
                    if (event.getType() == EventType.NodeChildrenChanged) {
                        log.info("------------ServerListWatcher:Node Child Changed Event Received");

                        addChildWatcher();
                    }
                }
            });

            String masterAddr = "";
            if (null != childs) {
                String masterNodeName = "";
                boolean isFrist = true;
                for (String child : childs) {

                    if (isFrist) {
                        masterNodeName = child;
                        isFrist = false;
                    } else {
                        if (masterNodeName.compareTo(child) > 0) {
                            masterNodeName = child;
                        }
                    }
                }

                log.info("------------ masterNode : " + masterNodeName);
                if (!"".equals(masterNodeName)) {
                    String masterPath = this.pathServer + "/" + masterNodeName;
                    if (masterPath.equals(registerAddressPath)) {
                        this.isMaster = true;
                    } else {
                        this.isMaster = false;
                    }

                    byte[] data = this.getZooKeeper().getData(masterPath, false, null);
                    masterAddr = new String(data, "utf-8");
                }

            }

            // 检查，如果所有注册的临时节点已经失效，重新连接zookeeper
            if (!StringUtils.isBlank(masterAddress) && (null == childs || childs.isEmpty())) {
                reRegisterJobTracker();
            } else {
                if (!masterAddress.equals(masterAddr) && null != listeners) {
                    masterAddress = masterAddr;
                    for (ClusterListener listener : listeners) {
                        listener.masterChanage(this.isMaster, masterAddr);
                    }
                }
            }

        } catch (KeeperException e) {
            e.printStackTrace();
            log.error("addChildWatcher failure", e);
        } catch (InterruptedException e) {
            e.printStackTrace();
            log.error("addChildWatcher failure", e);
        } catch (Exception e) {
            e.printStackTrace();
            log.error("addChildWatcher failure", e);
        }
    }

    public String getBlackList() {
        String blackList = null;
        try {
            if (null != blackListListener) {
                byte[] data = this.getZooKeeper().getData(this.pathBlackList, new Watcher() {

                    @Override public void process(WatchedEvent event) {
                        log.info("------------BlackListWatcher:BlackList Changed Event Received, EventType : " + event.getType().name());
                        if (EventType.NodeDataChanged == event.getType() || EventType.None == event.getType()) {
                            log.info("------------BlackListWatcher:BlackList Changed Event Received");

                            addBlackListWatcher();
                        }
                    }
                }, null);

                if (null != data) {
                    blackList = new String(data, "utf-8");
                }
            }

        }  catch (KeeperException e) {
            log.error("addBlackListWatcher failure", e);
        } catch (InterruptedException e) {
            log.error("addBlackListWatcher failure", e);
        } catch (Exception e) {
            log.error("addBlackListWatcher failure", e);
        }
        return blackList;
    }

    public boolean checkZKState() throws Exception {
        return zk != null && States.CONNECTED.equals(zk.getState());
    }

    public List<ACL> getAcl() {
        return acl;
    }

    public ZooKeeper getZooKeeper() throws Exception {
        if (this.checkZKState() == false) {
            reConnection();
        }
        return this.zk;
    }

    public String getRootPath() {
        return config.getRootPath();
    }

    public boolean isMaster() {
        return isMaster;
    }

    public String getPathServer() {
        return pathServer;
    }

    public void addListener(ClusterListener listener) {
        if (null != listener) {
            if (null == listeners) {
                listeners = new ArrayList<ClusterListener>();
            }
            listeners.add(listener);
        }
    }

    public void addBlackListListener(BlackListListener listener) {
        this.blackListListener = listener;
    }

    public static interface ClusterListener {

        void connected();

        void masterChanage(boolean isMaster, String masterAddress);

        void disconnected();

    }

    public static interface BlackListListener {

        void chanaged(String blankList);

    }

    protected void reRegisterJobTracker() {

    }

}
